package cn.hy.dataSync.task;


import cn.hy.dataSync.config.*;
import cn.hy.dataSync.job.State;
import cn.hy.dataSync.statistics.VMInfo;
import cn.hy.dataSync.statistics.communication.Communication;
import cn.hy.dataSync.statistics.communication.CommunicationTool;
import cn.hy.dataSync.statistics.container.communicator.AbstractContainerCommunicator;
import cn.hy.dataSync.task.runner.ReaderRunner;
import cn.hy.dataSync.task.runner.WriterRunner;
import cn.hy.dataSync.transport.channel.Channel;
import cn.hy.dataSync.transport.exchanger.RecordExchanger;
import cn.hy.dataSync.utils.ClassUtils;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.Validate;

import java.util.*;


/**
 * job  代表一次执行同步的一个实例
 * task 代表在这个job中执行任务的一个实例
 * 任务执行的容器
 */
@Slf4j
public class TaskContainer extends AbstractTaskContainer {
    /**
     * 所属jobId
     */
    private String jobId;

    /**
     * 当前taskId
     */
    private int taskId;

    /**
     * 使用的channel类
     */
    private String channelClazz;

    /**
     * task收集器使用的类
     */
    private String taskCollectorClazz;


    private TaskMonitor taskMonitor = TaskMonitor.getInstance();

    public TaskContainer(TaskConfig taskConfig,
                         AbstractContainerCommunicator containerCommunicator) {
        super(taskConfig, containerCommunicator);

        this.jobId = taskConfig.getJobConfig().getJobId();
        this.taskId = taskConfig.getTaskId();
        // 默认是内存
        this.channelClazz = taskConfig.getCoreConfig().getString(
                CoreConstant.TASK_CORE_TRANSPORT_CHANNEL_CLASS,
                CoreConstant.TASK_CORE_TRANSPORT_CHANNEL_DEFAULT);
        // TaskDirtyCollector 脏数据收集器
        this.taskCollectorClazz = taskConfig.getCoreConfig().getString(
                CoreConstant.TASK_CORE_STATISTICS_COLLECTOR_CLASS,
                CoreConstant.TASK_CORE_STATISTICS_COLLECTOR_DEFAULT);
    }

    public String getJobId() {
        return this.jobId;
    }

    public int getTaskId() {
        return this.taskId;
    }

    /**
     * 任务核心执行逻辑，执行一个任务
     */
    @Override
    public void start() {
        try {
            //状态 check 时间间隔较短，可以把任务及时分发到对应 channel 中
            int sleepIntervalInMillSec = this.getTaskConfig().getCoreConfig()
                    .getInteger(CoreConstant.TASK_SLEEP_INTERVAL, 100);

            //状态汇报时间间隔稍长，避免大量汇报
            long reportIntervalInMillSec = this.getTaskConfig().getCoreConfig()
                    .getLong(CoreConstant.TASK_REPORT_INTERVAL, 10000);

            //任务最大重试次数
            int taskMaxRetryTimes = this.getTaskConfig().getCoreConfig()
                    .getInteger(CoreConstant.TASK_FAILOVER_MAX_RETRY_TIMES, 1);

            //任务重试间隔时间
            long taskRetryIntervalInMsec = this.getTaskConfig().getCoreConfig()
                    .getLong(CoreConstant.TASK_FAILOVER_RETRY_INTERVAL_INMSEC, 10000);

            //任务重试最大等待时间
            long taskMaxWaitInMsec = this.getTaskConfig().getCoreConfig()
                    .getLong(CoreConstant.TASK_FAILOVER_MAX_WAIT_INMSEC, 60000);

            // 获取所有任务
            List<TaskConfig> taskConfigs = Lists.newArrayList(this.taskConfig);

            int taskCountInThisTaskGroup = taskConfigs.size();
            int channelNumber = Math.min(this.getTaskConfig().getJobConfig().getSpeedConfig()
                    .getChannel(), taskCountInThisTaskGroup);

            log.info(String.format("jobId [%s] taskId [%d] start", this.jobId, this.taskId));

            // 注册任务，初始化 containerCommunicator 的 key - taskId  value -> Communication map集合
            this.containerCommunicator.registerCommunication(taskConfigs);

            //taskId 与 task 配置
            Map<Integer, TaskConfig> taskConfigMap = buildTaskConfigMap(taskConfigs);

            //待运行 task 列表
            List<TaskConfig> taskQueue = buildRemainTasks(taskConfigs);

            //taskId 与上次失败实例
            Map<Integer, TaskExecutor> taskFailedExecutorMap = new HashMap<>();

            //正在运行 task
            List<TaskExecutor> runTasks = new ArrayList<>(channelNumber);

            //任务开始时间
            Map<Integer, Long> taskStartTimeMap = new HashMap<>();

            long lastReportTimeStamp = 0;
            Communication lastTaskGroupContainerCommunication = new Communication();

            while (true) {
                //1、判断 task 状态
                boolean failedOrKilled = false;
                // key - taskId  value -> Communication 汇报器
                Map<Integer, Communication> communicationMap = Maps.newHashMap();
                communicationMap.put(this.taskId, containerCommunicator.getCommunication(this.taskId));

                for (Map.Entry<Integer, Communication> entry : communicationMap.entrySet()) {
                    Integer taskId = entry.getKey();
                    Communication taskCommunication = entry.getValue();
                    if (!taskCommunication.isFinished()) {
                        continue;
                    }
                    TaskExecutor taskExecutor = removeTask(runTasks, taskId);

                    //上面从 runTasks 里移除了，因此对应在 monitor 里移除
                    taskMonitor.removeTask(taskId);

                    //失败，看 task 是否支持 failover，重试次数未超过最大限制
                    if (taskCommunication.getState() == State.FAILED) {
                        taskFailedExecutorMap.put(taskId, taskExecutor);
                        if (taskExecutor.getAttemptCount() < taskMaxRetryTimes) {
                            //关闭老的 executor
                            taskExecutor.shutdown();
                            //将 task 的状态重置
                            containerCommunicator.resetCommunication(taskId);
                            TaskConfig taskConfig = taskConfigMap.get(taskId);
                            //重新加入任务列表
                            taskQueue.add(taskConfig);
                        } else {
                            failedOrKilled = true;
                            break;
                        }
                    } else if (taskCommunication.getState() == State.KILLED) {
                        failedOrKilled = true;
                        break;
                    } else if (taskCommunication.getState() == State.SUCCEEDED) {
                        Long taskStartTime = taskStartTimeMap.get(taskId);
                        if (taskStartTime != null) {
                            Long usedTime = System.currentTimeMillis() - taskStartTime;
                            log.info("jobId[{}] taskId[{}] is succeed, used[{}]ms",
                                    this.jobId, taskId, usedTime);
                            taskStartTimeMap.remove(taskId);
                            taskConfigMap.remove(taskId);
                        }
                    }
                }

                //2、发现该 taskContainer 下 taskExecutor 的总状态失败则汇报错误
                if (failedOrKilled) {
                    lastTaskGroupContainerCommunication = reportTaskGroupCommunication(
                            lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);

                    throw new RuntimeException(lastTaskGroupContainerCommunication.getThrowable());
                }

                //3、有任务未执行，且正在运行的任务数小于最大通道限制
                //  执行任务
                Iterator<TaskConfig> iterator = taskQueue.iterator();
                // runTasks.size() < channelNumber  ? 为啥通道小于
                while (iterator.hasNext() && runTasks.size() < channelNumber) {
                    TaskConfig taskConfig = iterator.next();
                    Integer taskId = taskConfig.getTaskId();
                    int attemptCount = 1;
                    TaskExecutor lastExecutor = taskFailedExecutorMap.get(taskId);
                    if (lastExecutor != null) {
                        attemptCount = lastExecutor.getAttemptCount() + 1;
                        long now = System.currentTimeMillis();
                        long failedTime = lastExecutor.getTimeStamp();
                        //未到等待时间，继续留在队列
                        if (now - failedTime < taskRetryIntervalInMsec) {
                            continue;
                        }
                        //上次失败的task仍未结束
                        if (!lastExecutor.isShutdown()) {
                            if (now - failedTime > taskMaxWaitInMsec) {
                                markCommunicationFailed(taskId);
                                reportTaskGroupCommunication(lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);
                                throw new RuntimeException("task failover等待超时");
                            } else {
                                //再次尝试关闭
                                lastExecutor.shutdown();
                                continue;
                            }
                        } else {
                            log.info("jobId[{}] taskId[{}] attemptCount[{}] has already shutdown",
                                    this.jobId, taskId, lastExecutor.getAttemptCount());
                        }
                    }
                    // 具体获取任务执行
                    TaskConfig taskConfigForRun = taskMaxRetryTimes > 1 ? taskConfig.clone() : taskConfig;
                    TaskExecutor taskExecutor = new TaskExecutor(taskConfigForRun, attemptCount);
                    taskStartTimeMap.put(taskId, System.currentTimeMillis());
                    taskExecutor.doStart(); // 数据迁移核心执行逻辑

                    // 将调度的任务移除，并加入运行的任务
                    iterator.remove();
                    runTasks.add(taskExecutor);

                    //上面，增加 task 到 runTasks 列表，因此在 monitor 里注册。
                    taskMonitor.registerTask(taskId, this.containerCommunicator.getCommunication(taskId));

                    taskFailedExecutorMap.remove(taskId);
                    log.info("jobId[{}] taskId[{}] attemptCount[{}] is started", this.jobId, taskId, attemptCount);
                }

                //4、任务列表为空，executor 已结束, 搜集状态为success--->成功
                if (taskQueue.isEmpty() && isAllTaskDone(runTasks) && containerCommunicator.collectState() == State.SUCCEEDED) {
                    //成功的情况下，也需要汇报一次。否则在任务结束非常快的情况下，采集的信息将会不准确
                    lastTaskGroupContainerCommunication = reportTaskGroupCommunication(
                            lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);

                    log.info("jobId[{}] taskId[{}] completed", this.jobId, this.taskId);
                    break;
                }

                //5、如果当前时间已经超出汇报时间的 interval，那么我们需要马上汇报
                long now = System.currentTimeMillis();
                if (now - lastReportTimeStamp > reportIntervalInMillSec) {
                    lastTaskGroupContainerCommunication = reportTaskGroupCommunication(
                            lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);

                    lastReportTimeStamp = now;

                    //taskMonitor对于正在运行的task，每reportIntervalInMillSec进行检查
                    for (TaskExecutor taskExecutor : runTasks) {
                        taskMonitor.report(taskExecutor.getTaskId(),
                                this.containerCommunicator.getCommunication(taskExecutor.getTaskId()));
                    }
                }

                Thread.sleep(sleepIntervalInMillSec);
            }

            //6、最后还要汇报一次
            reportTaskGroupCommunication(lastTaskGroupContainerCommunication, taskCountInThisTaskGroup);
        } catch (Throwable e) {
            Communication nowTaskGroupContainerCommunication = this.containerCommunicator.getCommunication(this.taskId);

            if (nowTaskGroupContainerCommunication.getThrowable() == null) {
                nowTaskGroupContainerCommunication.setThrowable(e);
            }
            nowTaskGroupContainerCommunication.setState(State.FAILED);
            this.containerCommunicator.report(nowTaskGroupContainerCommunication);

            throw new RuntimeException(e);
        } finally {
            //最后打印 cpu 的平均消耗，GC 的统计
            VMInfo vmInfo = VMInfo.getVmInfo();
            if (vmInfo != null) {
                vmInfo.getDelta(false);
                log.info(vmInfo.totalString());
            }
        }
    }

    private Map<Integer, TaskConfig> buildTaskConfigMap(List<TaskConfig> taskConfigs) {
        Map<Integer, TaskConfig> map = new HashMap<>();
        for (TaskConfig taskConfig : taskConfigs) {
            map.put(taskConfig.getTaskId(), taskConfig);
        }
        return map;
    }

    private List<TaskConfig> buildRemainTasks(List<TaskConfig> taskConfigs) {
        return new LinkedList<>(taskConfigs);
    }

    private TaskExecutor removeTask(List<TaskExecutor> taskList, int taskId) {
        Iterator<TaskExecutor> iterator = taskList.iterator();
        while (iterator.hasNext()) {
            TaskExecutor taskExecutor = iterator.next();
            if (taskExecutor.getTaskId() == taskId) {
                iterator.remove();
                return taskExecutor;
            }
        }
        return null;
    }

    private boolean isAllTaskDone(List<TaskExecutor> taskList) {
        for (TaskExecutor taskExecutor : taskList) {
            if (!taskExecutor.isTaskFinished()) {
                return false;
            }
        }
        return true;
    }

    private Communication reportTaskGroupCommunication(Communication lastTaskGroupContainerCommunication,
                                                       int taskCount) {
        Communication nowTaskGroupContainerCommunication = this.containerCommunicator.getCommunication(this.taskId);
        nowTaskGroupContainerCommunication.setTimestamp(System.currentTimeMillis());
        Communication reportCommunication = CommunicationTool.getReportCommunication(nowTaskGroupContainerCommunication,
                lastTaskGroupContainerCommunication, taskCount);
        this.containerCommunicator.report(reportCommunication);
        return reportCommunication;
    }

    private void markCommunicationFailed(Integer taskId) {
        Communication communication = containerCommunicator.getCommunication(taskId);
        communication.setState(State.FAILED);
    }

    /**
     * TaskExecutor是一个完整task的执行器
     * 其中包括1：1的reader和writer
     */
    class TaskExecutor {
        private TaskConfig taskConfig;

        private final int taskId;

        private final int attemptCount;

        private final Thread readerThread;

        private final Thread writerThread;

        private final ReaderRunner readerRunner;

        private final WriterRunner writerRunner;

        /**
         * 该处的 taskCommunication 在多处用到：
         * 1. channel
         * 2. readerRunner 和 writerRunner
         * 3. reader 和 writer 的 taskCollector
         */
        private final Communication taskCommunication;

        public TaskExecutor(TaskConfig taskConfig, int attemptCount) {
            // 获取该 taskExecutor 的配置
            this.taskConfig = taskConfig;

            // 得到 taskId
            this.taskId = this.taskConfig.getTaskId();
            this.attemptCount = attemptCount;

            /*
             * 由 taskId 得到该 taskExecutor 的 Communication
             * 要传给 readerRunner 和 writerRunner，同时要传给 channel 作统计用
             */
            this.taskCommunication = containerCommunicator.getCommunication(taskId);
            Validate.notNull(this.taskCommunication, String.format("taskId[%d] 的 Communication 没有注册过", taskId));
            Channel channel = ClassUtils.instantiate(channelClazz, Channel.class, this.taskConfig); // 默认是内存
            channel.setCommunication(this.taskCommunication);

            /*
             * 脏数据搜集器
             */
            AbstractTaskDirtyCollector taskDirtyCollector;

            /*
             * 生成 writerThread
             */
            taskDirtyCollector = ClassUtils.instantiate(taskCollectorClazz,
                    AbstractTaskDirtyCollector.class, this.taskCommunication, ExpandType.WRITER);
            // 写任务的执行
            this.writerRunner = new WriterRunner(this.taskConfig);
            this.writerRunner.setRecordReceiver(new RecordExchanger(channel, this.taskCommunication, taskDirtyCollector));
            this.writerRunner.setRunnerCommunication(this.taskCommunication);
            this.writerRunner.setTaskDirtyCollector(taskDirtyCollector);
            this.writerThread = new Thread(this.writerRunner, String.format("%s-%d-writer", jobId, this.taskId));

            /*
             * 生成 readerThread
             */
            taskDirtyCollector = ClassUtils.instantiate(taskCollectorClazz,
                    AbstractTaskDirtyCollector.class, this.taskCommunication, ExpandType.READER);
            this.readerRunner = new ReaderRunner(this.taskConfig);
            this.readerRunner.setRecordSender(new RecordExchanger(channel, this.taskCommunication,
                    taskConfig.getJobConfig().getWriterConfig().getTransformers(), taskDirtyCollector));
            this.readerRunner.setRunnerCommunication(this.taskCommunication);
            this.readerRunner.setTaskDirtyCollector(taskDirtyCollector);
            this.readerThread = new Thread(this.readerRunner, String.format("%s-%d-reader", jobId, this.taskId));
        }

        public void doStart() {
            this.writerThread.start();

            // reader 没有起来，writer 不可能结束
            if (!this.writerThread.isAlive() || this.taskCommunication.getState() == State.FAILED) {
                throw new RuntimeException(this.taskCommunication.getThrowable());
            }

            this.readerThread.start();

            // 这里 reader 可能很快结束
            if (!this.readerThread.isAlive() && this.taskCommunication.getState() == State.FAILED) {
                // 这里有可能出现Reader线上启动即挂情况 对于这类情况 需要立刻抛出异常
                throw new RuntimeException(this.taskCommunication.getThrowable());
            }
        }

        // 检查任务是否结束
        private boolean isTaskFinished() {
            // 如果reader 或 writer没有完成工作，那么直接返回工作没有完成
            if (readerThread.isAlive() || writerThread.isAlive()) {
                return false;
            }

            return taskCommunication != null && taskCommunication.isFinished();
        }

        private int getTaskId() {
            return taskId;
        }

        private long getTimeStamp() {
            return taskCommunication.getTimestamp();
        }

        private int getAttemptCount() {
            return attemptCount;
        }

        private void shutdown() {
            writerRunner.shutdown();
            readerRunner.shutdown();
            if (writerThread.isAlive()) {
                writerThread.interrupt();
            }
            if (readerThread.isAlive()) {
                readerThread.interrupt();
            }
        }

        private boolean isShutdown() {
            return !readerThread.isAlive() && !writerThread.isAlive();
        }
    }
}
